package cn.xuqiudong.rpc.registry.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import cn.xuqiudong.rpc.cluster.exception.XqdCommonException;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 描述: 封装一些zk的基本操作
 * @author Vic.xu
 * @date 2022-02-16 10:01
 */
public class XqdZookeeperClient {

    private static Logger logger = LoggerFactory.getLogger(XqdZookeeperClient.class);

    private CuratorFramework client;

    /**
     * 缓存持久节点
     */
    private final Set<String> persistentExistNodePath = new HashSet<>();

    /**
     * 缓存监听器
     */
    private final ConcurrentMap<String, ConcurrentMap<ChildListener, XqdCuratorWatcherImpl>> childListeners = new ConcurrentHashMap<String, ConcurrentMap<ChildListener, XqdCuratorWatcherImpl>>();


    public XqdZookeeperClient(CuratorFramework client) {
        this.client = client;
    }

    /**
     * 删除路径
     * @param path 非临时路径
     */
    public void delete(String path) {
        persistentExistNodePath.remove(path);
        deletePath(path);
    }

    /**
     * 创建路径， 各级分别创建，父路径非临时
     * @param path 路径
     * @param ephemeral 是否临时
     */
    public void create(String path, boolean ephemeral) {
        if (!ephemeral) {
            if (persistentExistNodePath.contains(path)) {
                return;
            }
            if (checkExists(path)) {
                persistentExistNodePath.add(path);
                return;
            }
        }
        int i = path.lastIndexOf('/');
        if (i > 0) {
            create(path.substring(0, i), false);
        }
        if (ephemeral) {
            createEphemeral(path);
        } else {
            createPersistent(path);
            persistentExistNodePath.add(path);
        }
    }

    /**
     * 路径是否存在
     * @param path
     * @return
     */
    public boolean checkExists(String path) {
        try {
            if (client.checkExists().forPath(path) != null) {
                return true;
            }
        } catch (Exception e) {
        }
        return false;
    }

    /**
     * 创建持久路径
     * @param path
     */
    public void createPersistent(String path) {
        try {
            client.create().forPath(path);
        } catch (Exception e) {
            throw new XqdCommonException(e.getMessage(), e);
        }
    }

    /**
     * 创建临时节点
     * @param path
     */
    public void createEphemeral(String path) {
        try {
            if (checkExists(path)) {
                logger.info("临时节点" + path + "已存在，无需重新创建");
                return;
            }
                client.create().withMode(CreateMode.EPHEMERAL).forPath(path);

        } catch (Exception e) {
            throw new XqdCommonException(e.getMessage(), e);
        }
    }

    /**
     * 删除节点
     * @param path
     */
    protected void deletePath(String path) {
        try {
            client.delete().deletingChildrenIfNeeded().forPath(path);
        } catch (Exception e) {
            throw new XqdCommonException(e.getMessage(), e);
        }
    }

    /**
     * 查询子路径
     * @param path
     * @return
     */
    public List<String> getChildren(String path) {
        try {
            return client.getChildren().forPath(path);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 添加path的子节点的监听
     * @param path 当前节点
     * @param zkListener 监听器
     * @return
     */
    public List<String> addChildListener(String path, ChildListener zkListener) {
        // 从缓存中获取当前path对应的所有监听器，获取不到则新增
        ConcurrentMap<ChildListener, XqdCuratorWatcherImpl> listeners = childListeners.computeIfAbsent(path, k -> new ConcurrentHashMap<>());
        //获取当前ChildListener 对应的具体的Curator 监听器，不存在则新建
        XqdCuratorWatcherImpl targetListener = listeners.computeIfAbsent(zkListener, k -> new XqdCuratorWatcherImpl(client, zkListener, path));
        return addTargetChildListener(path, targetListener);
    }

    /**
     * 把监听添加到Curator
     * @param path 父节点2
     * @param listener 节点变化监听器
     */
    private List<String> addTargetChildListener(String path, XqdCuratorWatcherImpl listener) {
        try {
            return client.getChildren().usingWatcher(listener).forPath(path);
        } catch (Exception e) {
            throw new XqdCommonException(e.getMessage(), e);
        }
    }


    /**
     * 移除子节点的监听器， 先从当前缓存中查出监听器，然后把其关联的监视器置为null
     * @param path 节点
     * @param zkListener
     */
    public void removeChildListener(String path, ChildListener zkListener) {
        ConcurrentMap<ChildListener, XqdCuratorWatcherImpl> listeners = childListeners.get(path);
        if (listeners != null) {
            XqdCuratorWatcherImpl targetListener = listeners.remove(zkListener);
            if (targetListener != null) {
                removeTargetChildListener(path, targetListener);
            }
        }
    }

    public void removeTargetChildListener(String path, XqdCuratorWatcherImpl listener) {
        listener.unwatch();
    }
}
